package ETL

import (
	"context"
	"database/sql"
	"fmt"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/ref"
	"strings"
	"time"
)

type ReadOpt struct {
	SqlRows         *sql.Rows
	Schema          string
	Table           string
	TableColumnType []map[string]string
	DB              *sql.DB
	DBType          string
	Tx              *sql.Tx
	StopTime        time.Duration
	SqlLogBin       bool
	SqlMode         bool
	SessionSet      []string
	LobSwitch       bool
	FixRows         int64
	InsertPreFix    string
	Expect          string
	GisValFunc      func(any) any
}

// getColumnName 获取列名
func getColumnName(SqlRows *sql.Rows) (columns []string, err error) {
	var event = "[getColumnName]"
	if columns, err = SqlRows.Columns(); err != nil {
		err = ref.ErrAddPrintf(event, fmt.Errorf(fmt.Sprintf("DB Get the column fail. %v", err)))
		return
	}
	return
}
func manyColumnDataActive(gisVal func(any) any, columns []string, values []interface{}) (stringJoin []string, mapJoin map[string]any) {
	mapJoin = make(map[string]any)
	for i, k := range columns {
		var v interface{}
		val := values[i]
		b, ok := val.([]byte)
		if ok {
			v = string(b)
		} else {
			v = val
		}
		if gisVal != nil {
			v = gisVal(v)
		}
		mapJoin[k] = fmt.Sprintf("%v", v)
		stringJoin = append(stringJoin, fmt.Sprintf("%v", v))
	}
	return
}
func oneRowsActive(gisVal func(any) any, columns []string, values []interface{}, outPutType string) (res any) {
	valGarth, mapJoin := manyColumnDataActive(gisVal, columns, values)
	if len(valGarth) == 0 {
		return
	}
	switch outPutType {
	case String:
		return strings.Join(valGarth, DataColumnsSplit)
	case StringPoint:
		p := strings.Join(valGarth, DataColumnsSplit)
		return &p
	case StringPointGarth:
		var sliceJoin []*string
		for _, v := range valGarth {
			q := fmt.Sprintf("%v", v)
			sliceJoin = append(sliceJoin, &q)
		}
		return sliceJoin
	case Map:
		return mapJoin
	default:
		return
	}
}
func getSubExpectValue(expect string) (subExpect string) {
	switch expect {
	case String, StringGarth:
		subExpect = String
	case Map, SMap:
		subExpect = Map
	case StringPoint, StringPointGarth:
		subExpect = StringPoint
	case GarthStringPointGarth:
		subExpect = StringPointGarth
	}
	return
}
func subManyRowsActive(gisVal func(any) any, SqlRows *sql.Rows, columns []string, outPutType string) (manyRows []any, err error) {
	var event = "[subManyRowsActive]"
	// 定义一个切片，长度是字段的个数，切片里面的元素类型是sql.RawBytes
	valueStr := make([]interface{}, len(columns))
	//定义一个切片，元素类型是interface{}接口
	values := make([]interface{}, len(columns))
	for SqlRows.Next() {
		for i := 0; i < len(columns); i++ {
			valueStr[i] = &values[i]
		}
		if err = SqlRows.Scan(valueStr...); err != nil {
			err = ref.ErrAddPrintf(event, fmt.Errorf(fmt.Sprintf("DB Get the rows data fail. %v", err)))
			return
		}
		manyRows = append(manyRows, oneRowsActive(gisVal, columns, values, getSubExpectValue(outPutType)))
	}
	return
}

// manyRowsActive 多行数据进行处理
func manyRowsActive(gisVal func(any) any, SqlRows *sql.Rows, columns []string, outPutType string) (res any, err error) {
	var event = "[manyRowsActive]"
	var manyRowsData []any
	if manyRowsData, err = subManyRowsActive(gisVal, SqlRows, columns, outPutType); err != nil || manyRowsData == nil {
		return
	}
	switch outPutType {
	case String, StringPoint, Map:
		res = manyRowsData[0]
	case StringGarth:
		var qq []string
		for _, v := range manyRowsData {
			qq = append(qq, v.(string))
		}
		res = qq
	case StringPointGarth:
		var qq []*string
		for _, v := range manyRowsData {
			qq = append(qq, v.(*string))
		}
		res = qq
	case GarthStringPointGarth:
		var qq [][]*string
		for _, v := range manyRowsData {
			qq = append(qq, v.([]*string))
		}
		res = qq
	case SMap:
		var qq []map[string]any
		for _, v := range manyRowsData {
			qq = append(qq, v.(map[string]any))
		}
		res = qq
	default:
		err = ref.ErrAddPrintf(event, fmt.Errorf(fmt.Sprintf("type not match. curry type is %v", outPutType)))
		return
	}
	return
}

// rowsActiveMap 数据返回map
func baseRowsActiveMap(gisVal func(any) any, SqlRows *sql.Rows, expect string) (res any, err error) {
	var columns []string
	var event = "[rowsActiveMap]"
	if columns, err = getColumnName(SqlRows); err != nil {
		return nil, ref.ErrAddPrintf(event, err)
	}
	return manyRowsActive(gisVal, SqlRows, columns, expect)
}

func (r ReadOpt) chanRowsActiveS1(gisVal func(any) any, SqlRows *sql.Rows, _ string) (res any, err error) {
	var chanEntry = make(chan any, 100000)
	go func() {
		defer func() {
			close(chanEntry)
		}()
		var (
			columns    []string
			maxRetries = 1           // 最大重试次数
			retryDelay = time.Second // 重试延迟
		)
		for attempts := 0; attempts < maxRetries; attempts++ {
			if columns, err = getColumnName(SqlRows); err != nil {
				time.Sleep(retryDelay) // 等待后重试
				continue
			}
			// 检查是否有数据
			if !SqlRows.Next() {
				//err = fmt.Errorf(fmt.Sprintf("SqlRows.Next is nil"))
				return
			}
			valueStr := make([]interface{}, len(columns))
			values := make([]interface{}, len(columns))
			for i := 0; i < len(columns); i++ {
				valueStr[i] = &values[i]
			}
			if err = SqlRows.Scan(valueStr...); err != nil {
				time.Sleep(retryDelay) // 等待后重试
				continue
			}
			s1 := oneRowsActive(nil, columns, values, StringPointGarth)
			chanEntry <- s1
			// 处理余下行数据
			for SqlRows.Next() {
				for i := 0; i < len(columns); i++ {
					valueStr[i] = &values[i]
				}
				if err = SqlRows.Scan(valueStr...); err != nil {
					time.Sleep(retryDelay) // 等待后重试
					continue
				}
				s2 := oneRowsActive(gisVal, columns, values, StringPointGarth)
				chanEntry <- s2
			}
			if err = SqlRows.Close(); err != nil {
				err = fmt.Errorf(fmt.Sprintf("DB rows close fail. error is %v", err))
				return
			}
			return // 成功执行完毕
		}
	}()
	return chanEntry, err
}
func (r ReadOpt) expectFunc(rowsType string) (res func(gisVal func(any) any, SqlRows *sql.Rows, expect string) (res any, err error)) {
	switch rowsType {
	case Map, String, StringPoint, StringGarth, GarthStringPointGarth, SMap:
		return baseRowsActiveMap
	case FlowAny:
		return r.chanRowsActiveS1
	default:
		return
	}
}
func getRowsTx(conn *sql.Conn, strSql string) (rows *sql.Rows, err error) {
	return conn.QueryContext(context.TODO(), strSql)
	//return conn.QueryContext(context.Background(), strSql)
}

// sql语句循环执行，循环执行3次
func forExecQuerySql(conn *sql.Conn, strSql string) (rows *sql.Rows, err error) {
	// 执行包含可能导致错误的操作
	var event = "[forExecQuerySql]"
	defer func() {
		if re := ref.RecoverPanic(event, recover()); err == nil && re != nil {
			err = ref.ErrAddPrintf(event, re)
		}
	}()
	for i := 1; i < 4; i++ {
		if rows, err = getRowsTx(conn, strSql); err != nil {
			if i == 3 {
				err = ref.ErrAddPrintf(event, fmt.Errorf(fmt.Sprintf("exec sql fail. error info is {%v}", err)))
				return
			}
			time.Sleep(1 * time.Second)
		} else {
			return
		}
	}
	return
}

// SqlQuery sql语句查询 外部调用
func (r ReadOpt) sqlQuery(sessionOptions []string, strSql string, expect string) (execTime int64, entry any, err error) {
	var (
		rows  *sql.Rows
		event = "[SqlQuery]"
	)
	beginTime := time.Now()
	switch  {
	case strings.EqualFold(expect,FlowAny):
		if rows, err = r.DB.Query(strSql); err != nil {
			return
		}
	case len(sessionOptions) == 0:
		if rows, err = r.DB.Query(strSql); err != nil {
			return
		}
	default:
		var conn *sql.Conn
		// 从连接池获取连接
		if conn, err = r.DB.Conn(context.TODO()); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
		defer func() {
			if err = conn.Close(); err != nil {
				return
			}
		}()
		if err = setConnectionSessionVariables(conn, sessionOptions); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
		if rows, err = forExecQuerySql(conn, strSql); err != nil {
			err = ref.ErrAddPrintf(event, err)
			return
		}
	}
	defer func() {
		if expect != FlowAny {
			if err = rowsClose(rows); err != nil {
				err = ref.ErrAddPrintf(event, err)
				return
			}
		}
	}()
	if entry, err = r.expectFunc(expect)(r.GisValFunc, rows, expect); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	execTime = time.Since(beginTime).Milliseconds()
	return
}
func getSqlQueryFunc(r any, GisValFunc func(any) any, sessionOptions []string, strSql, expect string) (execTime int64, entry any, err error) {
	var event = "[getSqlQueryFunc]"
	switch r.(type) {
	case ReadOpt:
		p := r.(ReadOpt)
		if p.DB == nil {
			err = ref.ErrAddPrintf(event, fmt.Errorf(fmt.Sprintf("db connection is nil.")))
			return
		}
		p.GisValFunc = GisValFunc
		if execTime, entry, err = p.sqlQuery(sessionOptions, strSql, expect); err != nil {
			return
		}
		return
	case InstanceActiveS:
		if r.(InstanceActiveS).DB == nil {
			err = ref.ErrAddPrintf(event, fmt.Errorf(fmt.Sprintf("db connection is nil.")))
			return
		}
		return r.(InstanceActiveS).SqlQuery(strSql, expect)
	}
	return
}
func newExecSelectSql(s any, GisValFunc func(any) any, sessionOptions []string, sqlStr, expect string) (timeout int64, res any, err error) {
	var event = "[newExecSelectSql]"
	if timeout, res, err = getSqlQueryFunc(s, GisValFunc, sessionOptions, sqlStr, expect); err != nil {
		err = ref.ErrAddPrintf(event, err)
		return
	}
	return
}
func QuerySql(ctx context.Context, s any, GisValFunc func(any) any, sessionOptions, sqlStr []string, expect string) (result global.Return, err error) {
	defer func() {
		if err != nil {
			err = ref.ErrAddPrintf("QuerySql", err)
		}
	}()
	for _, v := range sqlStr {
		result.Sql = v
		if result.TimeOut, result.Result, err = newExecSelectSql(s, GisValFunc, sessionOptions, v, expect); err != nil || result.Result == nil {
			continue
		} else {
			log.MainLog().Debug("[QuerySql]", fmt.Sprintf("query sql exec is successful. sql is %v timeout is %v ", result.Sql, result.TimeOut))
			return
		}
	}
	return
}
